<?php
/**
 * @package PHPKit.
 * @author: mawenpei
 * @date: 2016/11/24
 * @time: 18:37
 */
namespace HuoKit\JobMan\Core;

use HuoKit\JobMan\Queue\BeanstalkdQueue;
use HuoKit\JobMan\Queue\IQueue;
use Psr\Log\LoggerInterface;
use swoole_process;

class QueueListener
{
    protected $workerName;

    protected $process;

    /**
     * @var IQueue
     */
    protected $queue;

    protected $logger;

    /**
     * @var ListenerStats
     */
    protected $stats;

    protected $times = 0;

    public function __construct($workerName, swoole_process $process,array $config,LoggerInterface $logger,ListenerStats $stats)
    {
        $this->workerName = $workerName;
        $this->process = $process;
        $this->config = $config;
        $this->logger = $logger;
        $this->stats = $stats;

        if(isset($config['queue'])){
            $this->queue = $config['queue'];
        }else{
            $this->queue = new BeanstalkdQueue($config);
        }
    }

    public function loop()
    {
        $worker = $this->createQueueWorker($this->workerName);
        while(true){
            $this->stats->touch($this->workerName,$this->process->pid,false,0);
            $stoping = $this->stats->isStoping();
            if ($stoping) {
                $this->logger->debug("process #{$this->process->pid} is exiting.");
                $this->process->exit(1);
                break;
            }

            $job = $this->reserveJob();

            if ($job===null) {
                $this->logger->debug('job is not exists');
                sleep(1);
                continue;
            }

            try {
                $result = $worker->execute($job);
                $this->stats->touch($this->workerName, $this->process->pid, false, 0);
            } catch(\Exception $e) {
                $message = sprintf('tube({$tubeName}, #%d): execute job #%d exception, `%s`', $this->process->pid, $job->getId(), $e->getMessage());
                $this->logger->error($message, ['data'=>$job->getData()]);
                continue;
            }

            $code = is_array($result) ? $result['code'] : $result;
            $this->finishJob($job, $result);
        }
    }

    private function reserveJob()
    {
        $job = $this->queue->pop($this->workerName);
        $this->times ++;

        $jobId = $job ? $job->getId() : 0;
        $jobData = $job ? $job->getData() : null;

        $this->stats->touch($this->workerName, $this->process->pid, true,$jobId);

        if ($job===null) {
            return null;
        }

        $this->logger->debug("worker ({$this->workerName}, #{$this->process->pid}): job #{$jobId} reserved.", ['data'=>$jobData]);

        return $job;
    }

    public function finishJob($job,$result)
    {
    }

    private function createQueueWorker($name)
    {
        $class = $this->config['workers'][$name]['class'];
        $worker = new $class($name, $this->config['workers'][$name]);
        $worker->setLogger($this->logger);
        return $worker;
    }
}

